package io.kiki.sba.registry.store.jdbc.repository.impl;


import io.kiki.sba.registry.store.api.meta.RecoverConfig;
import io.kiki.sba.registry.store.api.meta.RecoverConfigRepository;
import io.kiki.sba.registry.store.jdbc.domain.RecoverConfigDomain;
import io.kiki.sba.registry.store.jdbc.mapper.RecoverConfigMapper;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;


public class RecoverConfigJdbcRepository implements RecoverConfigRepository, ApplicationListener<ContextRefreshedEvent> {

    private static final Logger logger = LoggerFactory.getLogger(RecoverConfigJdbcRepository.class);
    private static final AtomicBoolean INIT_FINISH = new AtomicBoolean(false);
    private static final ConcurrentMap<String, RecoverConfig> callbackHandler = new ConcurrentHashMap<>();
    private static AtomicReference<Map<String, Set<String>>> CONFIG_MAP = new AtomicReference<>();
    private final ConfigWatcher watcher = new ConfigWatcher();

    @Autowired
    private RecoverConfigMapper recoverConfigMapper;

    @Autowired
    private TransactionTemplate transactionTemplate;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        ConcurrentUtils.createDaemonThread(this.getClass().getSimpleName() + "WatchDog", watcher).start();
    }

    @Override
    public Set<String> queryKey(String propertyTable) {
        if (StringUtils.isEmpty(propertyTable)) {
            throw new IllegalArgumentException("params is empty.");
        }
        Map<String, Set<String>> map = CONFIG_MAP.get();
        if (INIT_FINISH.get()) {
            return map.get(propertyTable);
        }
        List<RecoverConfigDomain> resp = recoverConfigMapper.query(propertyTable);
        if (CollectionUtils.isEmpty(resp)) {
            return Collections.emptySet();
        }
        return resp.stream().map(RecoverConfigDomain::getPropertyKey).collect(Collectors.toSet());
    }

    @Override
    public boolean save(String propertyTable, String propertyKey, String recoverClusterId) {
        if (StringUtils.isEmpty(propertyTable) || StringUtils.isEmpty(propertyKey)) {
            throw new IllegalArgumentException("params is empty.");
        }

        Set<String> keys = queryKey(propertyTable);
        if (!CollectionUtils.isEmpty(keys) && keys.contains(propertyKey)) {
            return true;
        }
        transactionTemplate.execute((status) -> {
            recoverConfigMapper.save(new RecoverConfigDomain(propertyTable, propertyKey));
            RecoverConfig recoverConfig = callbackHandler.get(propertyTable);
            if (recoverConfig != null) {
                recoverConfig.afterConfigSet(propertyKey, recoverClusterId);
            }
            return true;
        });

        return true;
    }

    @Override
    public boolean remove(String propertyTable, String propertyKey) {
        if (StringUtils.isEmpty(propertyTable) || StringUtils.isEmpty(propertyKey)) {
            throw new IllegalArgumentException("params is empty.");
        }
        recoverConfigMapper.remove(propertyTable, propertyKey);
        return true;
    }

    @Override
    public void waitSynced() {
        if (INIT_FINISH.get()) {
            return;
        }
        doRefresh();
    }

    @Override
    public void registerCallback(RecoverConfig config) {
        callbackHandler.put(config.tableName(), config);
    }

    public void doRefresh() {
        List<RecoverConfigDomain> result = recoverConfigMapper.queryAll();
        Map<String, Set<String>> configMap = new HashMap<>(result.size());
        for (RecoverConfigDomain config : result) {
            logger.info("[LOAD]config: {}", config);
            Set<String> set = configMap.computeIfAbsent(config.getPropertyTable(), k -> new HashSet<>());
            set.add(config.getPropertyKey());
        }
        this.CONFIG_MAP.set(configMap);
        logger.info("[LOAD-FINISH]CONFIG_MAP: {}", CONFIG_MAP.get());
        if (!INIT_FINISH.get()) {
            INIT_FINISH.set(true);
        }
    }

    class ConfigWatcher extends LoopExecuteTask {

        @Override
        public void _execute_() {
            doRefresh();
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(60, TimeUnit.SECONDS);
        }
    }
}
